package MaxTopn;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class MaxTopNMapper extends Mapper<LongWritable, Text, Text, MaxTopFlowBean>  {
    // 定义一个TreeMap作为存储数据的容器（按key排序）
    private TreeMap<MaxTopFlowBean, Text> flowMap = new TreeMap<>();
    private float sub;

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1. 获取一行
        String line = value.toString();

        // 2. 切割
        String[] fields = line.split(",");

        // 3. 封装数据
        String date = fields[0];
        String symbol = fields[1];
        float open = Float.parseFloat(fields[2]);
        float close = Float.parseFloat(fields[3]);
//        float low = Float.parseFloat(fields[4]);
//        float high = Float.parseFloat(fields[5]);
//        double volume = Double.parseDouble(fields[6]);
        sub = close - open;

//        MaxTopFlowBean flowBean = new MaxTopFlowBean(symbol, open, close, low, high, volume);
//        MaxTopFlowBean flowBean = new MaxTopFlowBean(volume);
        Text datetext = new Text(date);
        MaxTopFlowBean flowBean = new MaxTopFlowBean(open, close, sub, symbol);

        // 4. 向TreeMap中添加数据
        flowMap.put(flowBean, datetext);

        // 5. 限制TreeMap的数据量，超过10条就删除掉流量最小的一条数据
        if (flowMap.size() > 10) {
            flowMap.remove(flowMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 6 遍历treeMap集合，输出数据
        Iterator<MaxTopFlowBean> iterator = flowMap.keySet().iterator();

        while (iterator.hasNext()) {
            MaxTopFlowBean valueOut = iterator.next();
            Text keyOut = flowMap.get(valueOut);

            context.write(keyOut, valueOut);
        }
    }
}
